/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.wsinfra.wsrm.impl;

import java.util.Calendar;
import java.util.Hashtable;

import cgl.narada.wsinfra.WsMessageFlow;
import cgl.narada.wsinfra.WsProcessor;
import cgl.narada.wsinfra.exception.DeploymentException;
import cgl.narada.wsinfra.exception.IncorrectExchangeException;
import cgl.narada.wsinfra.exception.MessageFlowException;
import cgl.narada.wsinfra.exception.ParsingException;
import cgl.narada.wsinfra.exception.ProcessingException;
import cgl.narada.wsinfra.exception.UnknownExchangeException;
import cgl.narada.wsinfra.exception.WsFaultException;
import cgl.narada.wsinfra.schemas.addressing.EndpointReferenceType;
import cgl.narada.wsinfra.schemas.addressing.MessageIDDocument;
import cgl.narada.wsinfra.schemas.addressing.ToDocument;
import cgl.narada.wsinfra.schemas.soap.EnvelopeDocument;
import cgl.narada.wsinfra.schemas.wsrm.AckRequestedDocument;
import cgl.narada.wsinfra.schemas.wsrm.CreateSequenceResponseDocument;
import cgl.narada.wsinfra.schemas.wsrm.HandshakeType;
import cgl.narada.wsinfra.schemas.wsrm.SequenceAcknowledgementDocument;
import cgl.narada.wsinfra.schemas.wsrm.SequenceDocument;
import cgl.narada.wsinfra.wsa.AddressingHeaders;
import cgl.narada.wsinfra.wsa.ParseWsaHeaders;
import cgl.narada.wsinfra.wsa.WsaFaultRules;
import cgl.narada.wsinfra.wsa.WsaProcessingFactory;
import cgl.narada.wsinfra.wsrm.WsrmAckOperations;
import cgl.narada.wsinfra.wsrm.WsrmElementAddition;
import cgl.narada.wsinfra.wsrm.WsrmElementCreation;
import cgl.narada.wsinfra.wsrm.WsrmExchangeInfo;
import cgl.narada.wsinfra.wsrm.WsrmExchangeInfoCreator;
import cgl.narada.wsinfra.wsrm.WsrmFaults;
import cgl.narada.wsinfra.wsrm.WsrmNodeUtils;
import cgl.narada.wsinfra.wsrm.WsrmProcessingFactory;
import cgl.narada.wsinfra.wsrm.WsrmQNames;
import cgl.narada.wsinfra.wsrm.WsrmRequestCreator;
import cgl.narada.wsinfra.wsrm.WsrmSequencePolicies;
import cgl.narada.wsinfra.wsrm.WsrmSequencePolicyFactory;
import cgl.narada.wsinfra.wsrm.WsrmSource;
import cgl.narada.wsinfra.wsrm.storage.WsrmAuditStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmPolicyStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmProtocolStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmSequenceInfo;
import cgl.narada.wsinfra.wsrm.storage.WsrmSequenceInfoStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageEventsFactory;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageException;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageService;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageWidget;
/**
   This interfaces outlines the functionality of a WSRM sink.
   
   @author Shrideep Pallickara
   $Date: 2005/07/29 22:41:30 $
   $Revision: 1.12 $
*/


public class WsrmSourceNode extends WsProcessor implements WsrmSource {
  private EndpointReferenceType sourceEpr;
  private WsrmQNames wsrmQNames;
  private WsrmFaults wsrmFaults;
  private WsrmSequencePolicyFactory wsrmSequencePolicyFactory;  
  private WsrmExchangeInfoCreator wsrmExchangeInfoCreator;
  private WsrmAckOperations wsrmAckOperations;
  private WsrmElementAddition wsrmElementAddition;
  private WsrmElementCreation wsrmElementCreation;
  private WsrmRequestCreator wsrmRequestCreator;
  private WsrmNodeUtils wsrmNodeUtils;

  private WsrmStorageService wsrmStorageService;
  private WsrmProtocolStorageOperations wsrmProtocolOps;
  private WsrmSequenceInfoStorageOperations wsrmSequenceInfoOps;
  private WsrmPolicyStorageOperations wsrmPolicyOps;
  private WsrmAuditStorageOperations wsrmAuditOps;
  private WsrmStorageEventsFactory storageEventsFactory;

  private WsrmSourceNodeHelper sourceNodeHelper;

  private WsaFaultRules wsaFaultRules;
  private ParseWsaHeaders parseWsaHeaders; 

  private Hashtable syncObjects, createSequenceRequests;
  private WsMessageFlow wsMessageFlow;
  private String moduleName = "WsrmSourceNode: ";
  
  
  public WsrmSourceNode(String configInfo) throws DeploymentException {  
    try {
      wsrmStorageService = WsrmStorageService.getInstance(configInfo);
      wsrmProtocolOps = wsrmStorageService.getWsrmProtocolStorageOperations();
      wsrmSequenceInfoOps = 
	wsrmStorageService.getWsrmSequenceInfoStorageOperations(); 
      wsrmPolicyOps = wsrmStorageService.getWsrmPolicyStorageOperations();
      wsrmAuditOps =  wsrmStorageService.getWsrmAuditStorageOperations();
      
    } catch (WsrmStorageException wsrmStorageException) {
      String reason = moduleName + "Problems initializing storage services"
	+ ", please check the location of the config file [" + configInfo + 
	"] that was provided.\n" + wsrmStorageException.toString();
      throw new DeploymentException(reason);
    }
    
    storageEventsFactory = WsrmStorageEventsFactory.getInstance();
    
    wsrmQNames =  WsrmQNames.getInstance();
    wsrmFaults = WsrmProcessingFactory.getWsrmFaults();
    wsrmSequencePolicyFactory = 
      WsrmProcessingFactory.getWsrmSequencePolicyFactory();  
    wsrmExchangeInfoCreator = 
      WsrmProcessingFactory.getWsrmExchangeInfoCreator();
    wsrmAckOperations = WsrmProcessingFactory.getWsrmAckOperations();
    wsrmElementAddition = WsrmProcessingFactory.getWsrmElementAddition();
    wsrmElementCreation = WsrmProcessingFactory.getWsrmElementCreation();
    wsrmRequestCreator = WsrmProcessingFactory.getWsrmRequestCreator();
    wsrmNodeUtils = WsrmProcessingFactory.getWsrmNodeUtils();

    wsaFaultRules =  WsaProcessingFactory.getWsaFaultRules();
    parseWsaHeaders = WsaProcessingFactory.getParseWsaHeaders();

    sourceNodeHelper = WsrmSourceNodeHelper.getInstance();
    sourceNodeHelper.initialize(wsrmStorageService);

    syncObjects = new Hashtable();
    createSequenceRequests = new Hashtable();
  }
  

  /** Sets the message flow which the processor should use */
  public void setMessageFlow(WsMessageFlow wsMessageFlow) 
    throws DeploymentException {
    this.wsMessageFlow = wsMessageFlow;
    sourceNodeHelper.setMessageFlow(wsMessageFlow);
  }
  
  /** Gets the message flow which the processor should use. */
  public WsMessageFlow getMessageFlow() {
    return wsMessageFlow;
  }
  
  /** Process the exchange. The argument also indicates the direction in
      which the exchange has actually traversed. */
  public boolean processExchange(EnvelopeDocument envelopeDocument,
				 int direction)
    throws UnknownExchangeException, IncorrectExchangeException,
    ProcessingException, MessageFlowException {
    WsrmExchangeInfo wsrmExchangeInfo =
      wsrmExchangeInfoCreator.createWsrmExchangeInfo(envelopeDocument);
    AddressingHeaders addressingHeaders = 
      wsrmExchangeInfo.getAddressingHeaders();

    boolean isFault = isFault(addressingHeaders);
    if (isFault) {
      return true;
    }

    sourceNodeHelper.checkExchangeType(wsrmExchangeInfo, direction);
   
    try {
      /** Only valid exchanges to be processed now remain */
      if (direction == WsMessageFlow.FROM_APPLICATION) {
	if (wsrmExchangeInfo.isCreateSequence()) {
	  return true;
	}
	processMessageFromApplication(envelopeDocument, addressingHeaders);
	return true;
      }
      
      /** From here on every thing else has been received over the network */
      if (wsrmExchangeInfo.isCreateSequenceResponse()) {
	processCreateSequenceResponse(envelopeDocument, addressingHeaders);
	return false;
      }
      
      if (wsrmExchangeInfo.hasSequenceAcknowledgement()) {
	processSequenceAcknowledgement(envelopeDocument, addressingHeaders);
	return false;
      }

    } catch (WsFaultException wsFaultException) {
      EndpointReferenceType faultTo =
	wsaFaultRules.sendFaultTo(addressingHeaders);
      wsFaultException.setFaultTo(faultTo);
      manageFaultException(wsFaultException, direction);
    } catch (WsrmStorageException wsrmStorageException) {
      /** Need to manage storage exception here */
      System.out.println(moduleName + wsrmStorageException.toString() );
    }
    return true;
  }
  


  /** Process a create sequence response received from the sink */
  public void 
  processCreateSequenceResponse(EnvelopeDocument envelopeDocument,
				AddressingHeaders addressingHeaders) 
    throws WsFaultException, WsrmStorageException {
    CreateSequenceResponseDocument createSequenceResponseDocument 
      = wsrmNodeUtils.getCreateSequenceResponseDocument(envelopeDocument);
    
    HandshakeType createSequenceResponse = 
      createSequenceResponseDocument.getCreateSequenceResponse();
    
    String sequenceIdentifier = 
      createSequenceResponse.getIdentifier().getStringValue();
    
    String relatesTo = sourceNodeHelper.getRelatesTo(envelopeDocument);
    if (relatesTo == null) {
      String errorReport = "Trying to locate wsa:RelatesTo element in the " +
	"request, but the request DOES NOT contain this OR it's malformed.";  
      sourceNodeHelper.throwInvalidMessageFaultException(errorReport, 
							 addressingHeaders);
    }


    if (!createSequenceRequests.containsKey(relatesTo) ) {
      String errorReport = "Unable to determine if there was matching request"
	+ " to create this sequence. Please check the wsa:RelatesTo element " +
	"to see if it was populated correctly. The value that was read is [" +
	relatesTo +"]";
      sourceNodeHelper.throwInvalidMessageFaultException(errorReport, 
							 addressingHeaders);
    }
    
    
    String destination = (String) createSequenceRequests.get(relatesTo);
    if (wsrmSequenceInfoOps.hasSequenceIdForDestination(destination) ) {
      /** This is a duplicate create sequence response. A sequence identifier
	  under this name already exists for this destination */
      String errorReport ="This is a duplicate create sequence response. A " +
	"sequence identifier under this name already exists for destination ["
	+ destination + "]";
      sourceNodeHelper.throwInvalidMessageFaultException(errorReport, 
							 addressingHeaders);
    }
    
    wsrmSequenceInfoOps.
      addDestinationSequenceIdentifierPair(destination, 
					   sequenceIdentifier);
    
         
    /** Retrieve any policy related information that may be available within
	the envelope. If none exist, proceed to fill this information up
	using default policies that are available within the policyFactory */
    WsrmSequencePolicies wsrmSequencePolicies =
      sourceNodeHelper.getWsrmSequencePolicies(envelopeDocument,
					       sequenceIdentifier);
    WsrmSequenceInfo wsrmSequenceInfo =
      sourceNodeHelper.getWsrmSequenceInfo(addressingHeaders, 
					   sequenceIdentifier, relatesTo,
					   destination, wsrmSequencePolicies);
    /** Create a WsrmSequenceInfo object and store it */
    wsrmSequenceInfoOps.storeCreatedSequence(wsrmSequenceInfo);


    /** Proceed to release the pending wait */
    releasePendingWait(relatesTo);
  }







  
  
  /** Process sequence acknowledgements received over the wire from the sink */
  public void 
  processSequenceAcknowledgement(EnvelopeDocument envelopeDocument,
				 AddressingHeaders addressingHeaders)
    throws WsFaultException, WsrmStorageException, MessageFlowException {
    System.out.println(moduleName + "Processing Acknowledgements.");
    SequenceAcknowledgementDocument sequenceAckDocument = 
      wsrmNodeUtils.getSequenceAcknowledgement(envelopeDocument);
    SequenceAcknowledgementDocument.SequenceAcknowledgement acknowledgement =
      sequenceAckDocument.getSequenceAcknowledgement();
    String sequenceIdentifier = 
      acknowledgement.getIdentifier().getStringValue();
    
    long[] acknowledgements = 
      wsrmAckOperations.getAcknowledgements(sequenceAckDocument);
    
    if (acknowledgements != null) {
      sourceNodeHelper.checkForProblems(addressingHeaders, sequenceAckDocument,
					acknowledgements);
      /** Process the acknowledgements */
      wsrmProtocolOps.processAcknowledgementsOnSequence(sequenceIdentifier,
							acknowledgements);
    }
    
    long[] nacks = 
      sourceNodeHelper.getNegativeAcknowledgements(sequenceAckDocument);
    
    if (nacks != null) {
      sourceNodeHelper.checkForProblems(addressingHeaders, sequenceAckDocument,
					nacks);
    }

    /** Check for retransmissiona and if so, proceed to do handle that. */
    sourceNodeHelper.manageRetransmissions(sequenceIdentifier, nacks);
    
    /** Check to see if the sequence needs to be terminated. If so, proceed
	to create and issue the appropriate termination sequence exchange */
    sourceNodeHelper.processSequenceTermination(sequenceIdentifier);
  }
  






  /** Process exchange received from application, and send it across 
      reliably */
  public void 
  processMessageFromApplication(EnvelopeDocument envelopeDocument,
				AddressingHeaders addressingHeaders)
    throws WsFaultException, MessageFlowException, WsrmStorageException {
    sourceNodeHelper.checkForProblemsInToDocument(addressingHeaders);
    
    ToDocument toDocument = addressingHeaders.getTo();
    String destination = toDocument.getTo().getStringValue();
    
    if (!wsrmSequenceInfoOps.hasSequenceIdForDestination(destination) ) {
      String messageIdOfRequest = issueCreateSequenceRequest(toDocument);
      createSequenceRequests.put(messageIdOfRequest, destination);
      System.out.println(moduleName + "Issued a CreateSequence request with" +
			 " identifier [" + messageIdOfRequest +"]");
      initiateWaitOperation(messageIdOfRequest);
    }
    
    String sequenceIdentifier =
      wsrmSequenceInfoOps.getSequenceIdentifierForDestination(destination);
    WsrmSequenceInfo wsrmSequenceInfo = 
      wsrmSequenceInfoOps.getSequenceInfoUsingSequenceIdentifier(sequenceIdentifier);
    long previousMessageNumber = 
      wsrmProtocolOps.getMessageNumberLastAssigned(sequenceIdentifier);  
    
    sourceNodeHelper.checkForMessageRolloverFault(wsrmSequenceInfo, 
						  previousMessageNumber);
    
    boolean lastMessageOfSequence= 
      sourceNodeHelper.isLastMessageOfSequence(envelopeDocument);
    boolean ackRequested = 
      sourceNodeHelper.isAckRequested(envelopeDocument);
    
    /** Assign it a message number */
    long messageNumber = previousMessageNumber + 1;
    
    if (lastMessageOfSequence) {
      storageEventsFactory.setLastMessageNumberOnSequence(wsrmSequenceInfo,
							  messageNumber); 
      wsrmSequenceInfoOps.storeCreatedSequence(wsrmSequenceInfo);
    }

    Calendar expiresAt = 
      sourceNodeHelper.getSequenceExpiresAt(envelopeDocument);
    
    /** Add Sequence element to the message */
    SequenceDocument sequenceDocument = 
      wsrmElementCreation.newSequence(sequenceIdentifier, messageNumber,
				      lastMessageOfSequence, expiresAt);
    
    wsrmElementAddition.addSequence(envelopeDocument, sequenceDocument);
    
    if (lastMessageOfSequence) {
      ackRequested = true;
      storageEventsFactory.setLastMessageNumberOnSequence(wsrmSequenceInfo,
							  messageNumber); 
      wsrmSequenceInfoOps.storeCreatedSequence(wsrmSequenceInfo);
    }
    
    storeWsrmMessage(envelopeDocument, wsrmSequenceInfo, messageNumber);

    if (ackRequested) {
      AckRequestedDocument ackRequestedDocument;
      /** Add AckRequested element to the envelope document */
      if (wsrmSequenceInfo.hasLastMessageInfo() ) {
	long lastMessageNumber = wsrmSequenceInfo.getLastMessageNumber();
	ackRequestedDocument = 
	  wsrmElementCreation.newAckRequested(sequenceIdentifier,
					      lastMessageNumber);
      } else {
	ackRequestedDocument = 
	  wsrmElementCreation.newAckRequested(sequenceIdentifier);
      }
      wsrmElementAddition.addAckRequested(envelopeDocument, 
					  ackRequestedDocument);
    }
    
    /** Send the enevelope over the network. */
    //enrouteToNetwork(envelopeDocument);
  }


  private void storeWsrmMessage(EnvelopeDocument envelopeDocument, 
				WsrmSequenceInfo wsrmSequenceInfo,
				long messageNumber) 
    throws WsrmStorageException {
    
    if (wsrmSequenceInfo.getWsrmSequencePolicies() == null) {
      System.out.println(moduleName + "The WsrmSequencePolicies is NULL!");
    }

    WsrmSequencePolicies wsrmSequencePolicies = 
      wsrmSequenceInfo.getWsrmSequencePolicies();
    
    long retransmissionInterval = wsrmSequencePolicies.getRetransmissionInterval().getBaseRetransmissionInterval().getMilliseconds().longValue();
    retransmissionInterval += System.currentTimeMillis();
    
    String sequenceIdentifier = wsrmSequenceInfo.getSequenceIdentifier();
    
    WsrmStorageWidget wsrmStorageWidget = 
      storageEventsFactory.createWsrmStorageWidget(sequenceIdentifier,
						   messageNumber, 
						   false, 
						   retransmissionInterval,
						   envelopeDocument); 
    wsrmProtocolOps.store(wsrmStorageWidget);
  }
  
 

  /** This method issues a create sequence based on the specified toDocument.
      This method will also ensure that the request is sent over the network.
      The method returns the messageId of the created CreateSequenceRequest 
      that was issued. This can be used to be part of the wait-notify 
      scheme related to pausing operations up until the time that a response
      has been received over the wire. */
  private String issueCreateSequenceRequest(ToDocument toDocument) 
    throws MessageFlowException {
    String messageId = null;
    EnvelopeDocument envelopeDocument =
      wsrmRequestCreator.getCreateSequenceRequest(toDocument, sourceEpr);
    String errorReport = "issueCreateSequenceRequest->This SHOULD NOT HAVE " +
      "HAPPENED!, messageID problems should NOT arise!";
    try {
      MessageIDDocument messageIDDocument = 
	parseWsaHeaders.getMessageID(envelopeDocument);
      if (messageIDDocument == null) {
	reportError(errorReport);
	return messageId;
      }
      messageId = messageIDDocument.getMessageID().getStringValue();
    } catch (ParsingException parseEx) {
      reportError(errorReport);
      return messageId;
    }
    
    /** Send the enevelope over the network. */
    enrouteToNetwork(envelopeDocument);
    return messageId;
  }


  /** This method creates an Object that is used as the basis of a 
      wait()-notify() scheme. When a CreateSequenceResponse is received,
      the relatesTo element in that response (which will match this id) 
      will be used to relase the waiting. */
  private void initiateWaitOperation(String identifier) {
    Object syncObject = new Object();
    syncObjects.put(identifier, syncObject);
    
    synchronized (syncObject) {
      try {
	System.out.println(moduleName + "Waiting for CreateSequenceResponse");
	syncObject.wait();
	System.out.println(moduleName + "Received CreateSequenceResponse");
      } catch (InterruptedException ie) {
	reportError("Problems trying to wait on syncObject\n"); 
      }
    };
    
  }
  
  /** This method uses the relatesTo element within a CreateSequenceResponse 
      that is received to release a prior wait() operation that had been
      initaited. */
  private void releasePendingWait(String identifier) {
    Object syncObject = syncObjects.get(identifier);
    if (syncObject == null) {
      System.out.println(moduleName + "This identifer [" + identifier + 
			 "] is NOT a known identifier." +
			 " Can't do a notify() since there is no wait().");
      return;
    }
    
    synchronized (syncObject) {
      syncObject.notify();
    };
    
    syncObjects.remove(identifier);
  }

  
  /* Retrieves the endpoint reference associated with this node */
  public EndpointReferenceType getEndpointReference() {
    return sourceEpr;
  }


  /** Set the endpoint reference associated with this node */
  public void setEndpointReference(EndpointReferenceType endpointReference) {
    sourceEpr = endpointReference;
    sourceNodeHelper.setEndpointReference(sourceEpr);
  }

  private void reportError(String error) {
    System.out.println(moduleName + error);
  }

}



